package com.distributer.delayqueue.delayqueue.bucket;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.distributer.delayqueue.delayqueue.JobOperationService;
import com.distributer.delayqueue.delayqueue.support.DistributedLock;
import com.distributer.delayqueue.delayqueue.support.RedisQueueProperties;
import com.distributer.delayqueue.support.LifeCycle;
import com.distributer.delayqueue.util.NamedUtil;

public class BucketQueueManager implements LifeCycle{

	private static final Logger LOGGER = LoggerFactory.getLogger(BucketQueueManager.class);
	
	private static final String BUCKET_THREAD_NAME = "bucket-queue-thread-%s";
	
	private AtomicBoolean isRunning = new AtomicBoolean();
	private RedisQueueProperties redisQueueProperties;
	private boolean daemon = true;
	private Map<String, Timer> HOLD_TIMES = new ConcurrentHashMap<String,Timer>();
	private DistributedLock lock = null;
	private JobOperationService jobOperationService;
	
	@Override
	public void start() {
		int checkBucketSize = checkBucketSize();
		if (isRunning.compareAndSet(false, true)) {
			for (int i = 0; i < checkBucketSize; i++) {
				String bucketName = NamedUtil.buildBucketName(redisQueueProperties.getProfix(), redisQueueProperties.getName(), i);
				BucketTask bucketTask = new BucketTask(bucketName);
				bucketTask.setJobOperationService(jobOperationService);
				bucketTask.setRedisQueueProperties(redisQueueProperties);
				bucketTask.setLock(lock);
				String readyName = NamedUtil.buildRealTimeName(redisQueueProperties.getProfix(),redisQueueProperties.getName(), redisQueueProperties.getReadyName());
				bucketTask.setReadyName(readyName);
				String threadName = String.format(BUCKET_THREAD_NAME, i);
				Timer timer = new Timer(threadName, daemon);
				timer.schedule(bucketTask, 1000, redisQueueProperties.getBucketRoundRobinTime());
				HOLD_TIMES.put(threadName, timer);
				LOGGER.info(String.format("starting bucket thread: %s monitor bucketName: %s ", threadName,bucketName));
			}
		}
	}

	@Override
	public void stop() {
		if (isRunning.compareAndSet(true, false)) {
			for(Entry<String, Timer> entry : HOLD_TIMES.entrySet()){
				String threadName = entry.getKey();
                Timer  timer = entry.getValue();
                timer.cancel();
                LOGGER.info(String.format("stoping timer %s .....", threadName));
			}
		}
	}

	@Override
	public boolean isRunning() {
		return isRunning.get();
	}

	private int checkBucketSize(){
		if (redisQueueProperties.getBucketSize() <= 0) {
			return 1;
		}
		return redisQueueProperties.getBucketSize();
	}
	
	public void setRedisQueueProperties(RedisQueueProperties redisQueueProperties) {
		this.redisQueueProperties = redisQueueProperties;
	}

	public void setDaemon(boolean daemon) {
		this.daemon = daemon;
	}

	public void setLock(DistributedLock lock) {
		this.lock = lock;
	}
	
	public void setJobOperationService(JobOperationService jobOperationService) {
		this.jobOperationService = jobOperationService;
	}
	
}
